package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class MinStockWindowFunction extends RichWindowFunction<CleanBean, StockBean,String, TimeWindow> {
    /**
     * 开发步骤：
     *      1.新建MinStockWindowFunction 窗口函数
     *      2.初始化 MapState<String, StockBean>
     *      3.记录最新个股
     *      4.获取分时成交额和成交数量
     *      5.格式化日期
     *      6.封装输出数据
     *      7.更新MapState
     */

    //key:secCode  value:StockBean 用来缓存上一个窗口的分时行情数据
    private MapState<String,StockBean> stockMs=null;
    @Override
    public void open(Configuration parameters) throws Exception {
        stockMs = getRuntimeContext().getMapState(
                        new MapStateDescriptor<String, StockBean>("stockMs",
                        String.class, StockBean.class));
    }

    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<CleanBean> iterable, Collector<StockBean> collector) throws Exception {
        //todo 3.记录最新个股
        CleanBean cleanBean = null;
        for (CleanBean line : iterable) {
            if(cleanBean==null){
                cleanBean=line;
            }
            if(line.getEventTime()>cleanBean.getEventTime()){
                cleanBean=line;
            }
        }

        //todo 4.获取分时成交额和成交数量
        //分时成交量   (当前分钟的总成交量-上一分钟的总成交量）
        //分时成交金额 (当前分钟的总成交金额-上一分钟的总成交金额）
        StockBean stockBeanLast = stockMs.get(cleanBean.getSecCode());
        Long curTradeVol = 0L; //分时成交量
        Long curTradeAmt = 0L; //分时成交金额
        if(stockBeanLast!=null){
            curTradeVol =  cleanBean.getTradeVolumn() - stockBeanLast.getTradeVolDay();
            curTradeAmt =  cleanBean.getTradeAmt() - stockBeanLast.getTradeAmtDay();
            System.out.println("分时成交量："+curTradeVol+"  分时成交金额："+curTradeAmt);
        }
        //todo 5.格式化日期
        Long tradeTime = DateUtil.longTimeTransfer(cleanBean.getEventTime(), Constant.format_YYYYMMDDHHMMSS);
        //todo 6.封装输出数据
        StockBean stockBean = new StockBean();
        stockBean.setEventTime(cleanBean.getEventTime());
        stockBean.setSecCode(cleanBean.getSecCode());
        stockBean.setSecName(cleanBean.getSecName());
        stockBean.setPreClosePrice(cleanBean.getPreClosePrice());
        stockBean.setOpenPrice(cleanBean.getOpenPrice());
        stockBean.setClosePrice(cleanBean.getTradePrice());
        stockBean.setHighPrice(cleanBean.getMaxPrice());
        stockBean.setLowPrice(cleanBean.getMinPrice());
        stockBean.setTradeVol(curTradeVol);
        stockBean.setTradeAmt(curTradeAmt);
        stockBean.setTradeVolDay(cleanBean.getTradeVolumn());
        stockBean.setTradeAmtDay(cleanBean.getTradeAmt());
        stockBean.setTradeTime(tradeTime); //20220214121733
        stockBean.setSource(cleanBean.getSource());
        //todo 7.更新MapState
        stockMs.put(stockBean.getSecCode(),stockBean);

        //todo 8.收集数据
        collector.collect(stockBean);
    }
}
